package Creating_Observables

import (
	"fmt"
	"github.com/reactivex/rxgo/v2"
	"time"
)

// 创建一个热可观察到的一个通道。
// 对象一旦观察被创建之后就会被消费。
//一个观察者只能看到那一刻起他订阅了可观测的对象。
func Demo_FromEventSource() {
	ch := make(chan rxgo.Item)
	observable := rxgo.FromEventSource(ch)
	go func() {
		for i := 0; i < 8; i++ {
			time.Sleep(time.Second)
			ch <- rxgo.Of(i)
			fmt.Println("生产：", i)
		}
		close(ch)
	}()

	// 订阅 1
	go func() {
		time.Sleep(time.Second * 3)
		for item := range observable.Observe() {
			if item.Error() {
				fmt.Println("订阅 1：", item.E)
			}
			fmt.Println("订阅 1：", item.V)
		}
	}()
	time.Sleep(time.Second * 6)
	// 订阅 2
	for item := range observable.Observe() {
		if item.Error() {
			fmt.Println("订阅 2：", item.E)
		}
		fmt.Println("订阅 2：", item.V)
	}
}

// 使用背压时，rxgo.Block参数 阻塞生产，保证每一个对象均被消费者消费
// rxgo.Drop参数 丢弃对象 仅当前产出订阅的数据
func Demo_FromEventSourceWithBackPressureStrategy() {
	ch := make(chan rxgo.Item)
	observable := rxgo.FromEventSource(ch,
		rxgo.WithBackPressureStrategy(rxgo.Drop),
		rxgo.WithBufferedChannel(10))
	go func() {
		for i := 0; i < 8; i++ {
			time.Sleep(time.Millisecond * 200)
			ch <- rxgo.Of(i)
			fmt.Println("生产：", i)
		}
		time.Sleep(time.Second * 10)
		close(ch)
	}()

	// 订阅 1
	go func() {
		time.Sleep(time.Second)
		for item := range observable.Observe() {
			time.Sleep(time.Second * 2)
			if item.Error() {
				fmt.Println("订阅 1：", item.E)
			}
			fmt.Println("订阅 1：", item.V)
		}
	}()
	time.Sleep(time.Second)
	// 订阅 2
	for item := range observable.Observe() {
		time.Sleep(time.Second * 4)
		if item.Error() {
			fmt.Println("订阅 2：", item.E)
		}
		fmt.Println("订阅 2：", item.V)
	}
}
